package cn.azzhu.day08

import cn.azzhu.day02.SensorSource
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api._
import org.apache.flink.api.scala._
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction}
import org.apache.flink.types.Row
/**
 * Flink-Table-SQL：聚合函数和表聚合函数
 * @author azzhu
 * @create 2020-09-23 23:43:54
 */
object AggregateFunctionExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tEnv = StreamTableEnvironment.create(env, settings)

    val stream = env.addSource(new SensorSource)

    val avgTemp = new AvgTemp

    //table api
    val table = tEnv.fromDataStream(stream, 'id, 'timestamp as 'ts, 'temperature as 'temp)

    table
      .groupBy('id)
      .aggregate(avgTemp('temp) as 'avgTemp)
      .select('id,'avgTemp)
      .toRetractStream[Row]
     // .print()

    //sql写法
    tEnv.createTemporaryView("sensor",table)
    tEnv.registerFunction("avgTemp",avgTemp)

    tEnv
      .sqlQuery("select id,avgTemp(temp) from sensor group by id")
      .toRetractStream[Row]
      .print()

    env.execute("AggregateFunctionExample")
  }

  //累加器的类型
  class AvgTempAcc {
    var sum:Double = 0.0
    var count:Int = 0
  }

  //第一个泛型：温度值的类型
  //第二个泛型：累加器的类型
  class AvgTemp extends AggregateFunction[Double,AvgTempAcc] {
    override def getValue(acc: AvgTempAcc): Double = {
      acc.sum / acc.count
    }

    //创建累加器
    override def createAccumulator(): AvgTempAcc = new AvgTempAcc

    //累加规则
    def accumulate(acc:AvgTempAcc,temp:Double):Unit = {
      acc.sum += temp
      acc.count += 1
    }
  }
}
